package jupitermouse.site.source.util;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Properties;
import java.util.Random;

/**
 * @author qiqiang.ren@hand-china.com
 */
@Slf4j
public class MyKafkaProducer {
    public static void main(String[] args) throws InterruptedException {
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "192.168.81.130:9092");
        properties.setProperty("zookeeper.connect", "192.168.81.130:2181");
        properties.setProperty("key.serializer", StringSerializer.class.getName());
        properties.setProperty("value.serializer", StringSerializer.class.getName());

        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);

        String topic = "kafka-source";
        //通过Kafka的Broker生产数据
        for (int i = 0; i < 1000; i++) {
            StringBuilder builder = new StringBuilder();
            builder.append("hand" + i).append(" ");
            log.info("string:{}", builder.toString());
            producer.send(new ProducerRecord<>(topic, builder.toString()));
            Thread.sleep(1000);
        }

    }

}
